package benchmarks.tpch.mysql;

import api.Configuration;
import api.result.Result;
import api.connection.ConnectionMySQL;
import api.connection.DBConnection;
import api.result.ResultKV;
import api.result.UnitResult;
import benchmarks.tpch.TPCHLoader;
import benchmarks.tpch.utils.DBtool;
import benchmarks.tpch.utils.Types;
import com.google.gson.Gson;
import static benchmarks.tpch.utils.Types.*;

import java.io.*;
import java.sql.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class TPCHLoaderMySQL extends TPCHLoader {
    private DBConnection<Connection> dbConnection;
    private Connection conn;
    private final static int configCommitCount = 10000; // commit every n records

    public Result loadData(DBConnection connection) {
        DBtool.prepareTBL(getScaleFactor());
        long start = System.currentTimeMillis();
        this.dbConnection = connection;
        this.conn = dbConnection.makeConnection();
        System.out.println("start create database");
        createDatabase();
        try {
            System.out.println("start load tables");
            loadTBL();
        } catch (SQLException e) {
            System.out.println("loading tpc-h on mysql failed");
            e.printStackTrace();
        }
        System.out.println("start build index");
        buildIndex();
        long end = System.currentTimeMillis();
        ResultKV<UnitResult> resTime = new ResultKV<>(
                "tpc-h\tMySQL\t" + getScaleFactor() + "\tloadData",
                new UnitResult(end - start));
        ResultKV<UnitResult> resRecord = new ResultKV<>(
                "tpc-h\tMySQL\t" + getScaleFactor() + "\ttotalRecords",
                new UnitResult(totalRows, "records"));
        ResultKV<UnitResult> resSize = null;
        ResultKV<UnitResult> resIndex = null;
        try {
            ResultSet status = this.conn.createStatement().executeQuery("SELECT sum(data_length) AS data_size, sum(index_length) AS index_size\n" +
                    "FROM information_schema.TABLES WHERE table_schema = \"tpch\";");
            status.first();
            resSize = new ResultKV<>(
                    "tpc-h\tMySQL\t" + getScaleFactor() + "\tdataSize",
                    new UnitResult(status.getLong("data_size") >> 10, "kb"));
            resIndex = new ResultKV<>(
                    "tpc-h\tMySQL\t" + getScaleFactor() + "\tindexSize",
                    new UnitResult(status.getLong("index_size") >> 10, "kb"));
        } catch (SQLException e) {
            e.printStackTrace();
        }
        try {
            this.conn.close();
        } catch (SQLException e) {
            e.printStackTrace();
        }
        System.out.println("loading tpc-h on mysql ok");
        return resTime.union(resRecord).union(resSize).union(resIndex);
    }

    private static enum StmtTypes { CUSTOMER, LINEITEM, NATION, ORDERS,
                                    PART, PARTSUPP, REGION, SUPPLIER}

//    private static PreparedStatement customerPrepStmt;
//    private static PreparedStatement lineitemPrepStmt;
//    private static PreparedStatement nationPrepStmt;
//    private static PreparedStatement ordersPrepStmt;
//    private static PreparedStatement partPrepStmt;
//    private static PreparedStatement partsuppPrepStmt;
//    private static PreparedStatement regionPrepStmt;
//    private static PreparedStatement supplierPrepStmt;

    public TPCHLoaderMySQL(double scaleFactor) {
        super(scaleFactor);
    }

    private void createDatabase() {
        try {
            ScriptRunner scriptRunner = new ScriptRunner(conn, false, true);
            scriptRunner.runScript(new BufferedReader(new InputStreamReader(
                    this.getClass().getClassLoader().getResourceAsStream("tpch-mysql-ddl.sql"))));
            } catch (IOException e) {
            e.printStackTrace();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    private void buildIndex() {
        try {
            ScriptRunner scriptRunner = new ScriptRunner(conn, false, true);
            scriptRunner.runScript(new BufferedReader(new InputStreamReader(
                    this.getClass().getClassLoader().getResourceAsStream("tpch-mysql-index-ddl.sql"))));
        } catch (IOException e) {
            e.printStackTrace();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    private PreparedStatement makePrepareStatement(Connection conn, StmtTypes stmtTypes) throws SQLException{
        PreparedStatement stmt = null;
        try {
            switch (stmtTypes) {
                case CUSTOMER:
                    stmt = conn.prepareStatement("INSERT INTO customer "
                            + "(c_custkey, c_name, c_address, c_nationkey,"
                            + " c_phone, c_acctbal, c_mktsegment, c_comment ) "
                            + "VALUES (?, ?, ?, ?, ?, ?, ?, ?)");
                    break;
                case LINEITEM:
                    stmt = conn.prepareStatement("INSERT INTO lineitem "
                            + "(l_orderkey, l_partkey, l_suppkey, l_linenumber,"
                            + " l_quantity, l_extendedprice, l_discount, l_tax,"
                            + " l_returnflag, l_linestatus, l_shipdate, l_commitdate,"
                            + " l_receiptdate, l_shipinstruct, l_shipmode, l_comment) "
                            + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)");
                    break;
                case NATION:
                    stmt = conn.prepareStatement("INSERT INTO nation "
                            + "(n_nationkey, n_name, n_regionkey, n_comment) "
                            + "VALUES (?, ?, ?, ?)");
                    break;
                case ORDERS:
                    stmt = conn.prepareStatement("INSERT INTO orders "
                            + "(o_orderkey, o_custkey, o_orderstatus, o_totalprice,"
                            + " o_orderdate, o_orderpriority, o_clerk, o_shippriority,"
                            + " o_comment) "
                            + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)");
                    break;
                case PART:
                    stmt = conn.prepareStatement("INSERT INTO part "
                            + "(p_partkey, p_name, p_mfgr, p_brand, p_type,"
                            + " p_size, p_container, p_retailprice, p_comment) "
                            + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)");
                    break;
                case PARTSUPP:
                    stmt = conn.prepareStatement("INSERT INTO partsupp "
                            + "(ps_partkey, ps_suppkey, ps_availqty, ps_supplycost,"
                            + " ps_comment) "
                            + "VALUES (?, ?, ?, ?, ?)");
                    break;
                case REGION:
                    stmt = conn.prepareStatement("INSERT INTO region "
                            + " (r_regionkey, r_name, r_comment) "
                            + "VALUES (?, ?, ?)");

                    break;
                case SUPPLIER:
                    stmt = conn.prepareStatement("INSERT INTO supplier "
                            + "(s_suppkey, s_name, s_address, s_nationkey, s_phone,"
                            + " s_acctbal, s_comment) "
                            + "VALUES (?, ?, ?, ?, ?, ?, ?)");
                    break;
                default:
                        throw new RuntimeException();
            }
        } catch (SQLException e) {
            conn.rollback();
        }
        return stmt;
    }

    private void loadTBL() throws SQLException{
//        try {
//            customerPrepStmt = conn.prepareStatement("INSERT INTO customer "
//                    + "(c_custkey, c_name, c_address, c_nationkey,"
//                    + " c_phone, c_acctbal, c_mktsegment, c_comment ) "
//                    + "VALUES (?, ?, ?, ?, ?, ?, ?, ?)");
//
//            lineitemPrepStmt = conn.prepareStatement("INSERT INTO lineitem "
//                    + "(l_orderkey, l_partkey, l_suppkey, l_linenumber,"
//                    + " l_quantity, l_extendedprice, l_discount, l_tax,"
//                    + " l_returnflag, l_linestatus, l_shipdate, l_commitdate,"
//                    + " l_receiptdate, l_shipinstruct, l_shipmode, l_comment) "
//                    + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)");
//
//            nationPrepStmt = conn.prepareStatement("INSERT INTO nation "
//                    + "(n_nationkey, n_name, n_regionkey, n_comment) "
//                    + "VALUES (?, ?, ?, ?)");
//
//            ordersPrepStmt = conn.prepareStatement("INSERT INTO orders "
//                    + "(o_orderkey, o_custkey, o_orderstatus, o_totalprice,"
//                    + " o_orderdate, o_orderpriority, o_clerk, o_shippriority,"
//                    + " o_comment) "
//                    + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)");
//
//            partPrepStmt = conn.prepareStatement("INSERT INTO part "
//                    + "(p_partkey, p_name, p_mfgr, p_brand, p_type,"
//                    + " p_size, p_container, p_retailprice, p_comment) "
//                    + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)");
//
//            partsuppPrepStmt = conn.prepareStatement("INSERT INTO partsupp "
//                    + "(ps_partkey, ps_suppkey, ps_availqty, ps_supplycost,"
//                    + " ps_comment) "
//                    + "VALUES (?, ?, ?, ?, ?)");
//
//            regionPrepStmt = conn.prepareStatement("INSERT INTO region "
//                    + " (r_regionkey, r_name, r_comment) "
//                    + "VALUES (?, ?, ?)");
//
//            supplierPrepStmt = conn.prepareStatement("INSERT INTO supplier "
//                    + "(s_suppkey, s_name, s_address, s_nationkey, s_phone,"
//                    + " s_acctbal, s_comment) "
//                    + "VALUES (?, ?, ?, ?, ?, ?, ?)");
//
//        } catch (SQLException se) {
////            LOG.debug(se.getMessage());
//            conn.rollback();
//
//        } catch (Exception e) {
//            e.printStackTrace();
//            conn.rollback();
//        } // end try

//        conn.setAutoCommit(false);
//        loadHelper();
//        conn.commit();

        Thread loaders[] = new Thread[8];
        loaders[0] = loadCustomers();
        loaders[1] = loadLineItems();
        loaders[2] = loadNations();
        loaders[3] = loadOrders();
        loaders[4] = loadParts();
        loaders[5] = loadPartSupps();
        loaders[6] = loadRegions();
        loaders[7] = loadSuppliers();

        for (int i = 0; i < 8; ++i)
            if (loaders[i] != null)
                loaders[i].start();

        for (int i = 0; i < 8; ++i) {
            try {
                if (loaders[i] != null)
                    loaders[i].join();
            } catch(InterruptedException e) {
                e.printStackTrace();
//                LOG.error(e.getMessage());
            }
        }
    }

    Thread loadCustomers() {
        return new Thread(new TableLoader("Customer", customerTypes, StmtTypes.CUSTOMER, this, dbConnection));
    }

    Thread loadLineItems() {
        return new Thread(new TableLoader("LineItem", lineitemTypes, StmtTypes.LINEITEM, this, dbConnection));
    }

    Thread loadNations() {
        return new Thread(new TableLoader("Nation", nationTypes, StmtTypes.NATION, this, dbConnection));
    }

    Thread loadOrders() {
        return new Thread(new TableLoader("Orders", ordersTypes, StmtTypes.ORDERS, this, dbConnection));
    }

    Thread loadParts() {
        return new Thread(new TableLoader("Part", partTypes, StmtTypes.PART, this, dbConnection));
    }

    Thread loadPartSupps() {
        return new Thread(new TableLoader("PartSupp", partsuppTypes, StmtTypes.PARTSUPP, this, dbConnection));
    }

    Thread loadRegions() {
        return new Thread(new TableLoader("Region", regionTypes, StmtTypes.REGION, this, dbConnection));
    }

    Thread loadSuppliers() {
        return new Thread(new TableLoader("Supplier", supplierTypes, StmtTypes.SUPPLIER, this, dbConnection));
    }

    protected long totalRows = 0;

    protected long loadHelper() {
        Thread loaders[] = new Thread[8];
        loaders[0] = loadCustomers();
        loaders[1] = loadLineItems();
        loaders[2] = loadNations();
        loaders[3] = loadOrders();
        loaders[4] = loadParts();
        loaders[5] = loadPartSupps();
        loaders[6] = loadRegions();
        loaders[7] = loadSuppliers();

        for (int i = 0; i < 8; ++i)
            if (loaders[i] != null)
                loaders[i].start();

        for (int i = 0; i < 8; ++i) {
            try {
                if (loaders[i] != null)
                    loaders[i].join();
            } catch(InterruptedException e) {
//                LOG.error(e.getMessage());
            }
        }

        return this.totalRows;
    }

    private class TableLoader implements Runnable {
        String tableName;
        TPCHLoaderMySQL parent;
        Types.CastTypes[] types;
        StmtTypes stmtTypes;
        PreparedStatement prepStmt;
        final DBConnection<Connection> dbConnection;

        private Connection conn;

        TableLoader(String tableName, CastTypes[] types, StmtTypes stmtTypes
                , TPCHLoaderMySQL parent, DBConnection<Connection> dbConnection) {
            this.tableName = tableName;
            this.parent = parent;
            this.types = types;
            this.stmtTypes = stmtTypes;
            this.dbConnection = dbConnection;
        }

        @Override
        public void run() {
            BufferedReader bufferedReader = null;
            int recordsRead = 0;

            try {
                conn = dbConnection.makeConnection();
                conn.setAutoCommit(false);
                prepStmt = makePrepareStatement(conn, stmtTypes);
                try {
                    File file = new File(DBtool.getDBpath(getScaleFactor())
                            , tableName.toLowerCase() + ".tbl");
                    bufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream(file)));
                    String line;
                    while ((line = bufferedReader.readLine()) != null) {
                        String[] fields = line.split("\\|");
                        for (int i = 0; i < types.length; i++) {
                            String field = fields[i];
                            switch (types[i]) {
                                case LONG:
                                    prepStmt.setLong(i + 1, Long.parseLong(field));
                                    break;
                                case DOUBLE:
                                    prepStmt.setDouble(i + 1, Double.parseDouble(field));
                                    break;
                                case STRING:
                                    prepStmt.setString(i + 1, field);
                                    break;
                                case DATE:
                                    // yyyy-mm-dd
                                    Pattern isoFmt = Pattern.compile("^\\s*(\\d{4})-(\\d{2})-(\\d{2})\\s*$");
                                    Matcher isoMatcher = isoFmt.matcher(field);
                                    java.sql.Date fieldAsDate = null;
                                    if (isoMatcher.find()) {
                                        fieldAsDate = new java.sql.Date(
                                                Integer.parseInt(isoMatcher.group(1)) - 1900,
                                                Integer.parseInt(isoMatcher.group(2)),
                                                Integer.parseInt(isoMatcher.group(3)));
                                    } else {
                                        throw new RuntimeException("Unrecognized date \""
                                                + field + "\" in TBL file: "
                                                + file.getAbsolutePath());
                                    }
                                    prepStmt.setDate(i + 1, fieldAsDate);
                                    break;
                                default:
                                    throw new RuntimeException("Unrecognized type for prepared statement");
                            }
                        }
                        prepStmt.addBatch();
                        recordsRead++;
                        if ((recordsRead % configCommitCount) == 0) {
                            prepStmt.executeBatch();
                            prepStmt.clearBatch();
                            conn.commit();
                        }
                    }
                    prepStmt.executeBatch();
                    conn.commit();
                } catch (SQLException e) {
                    conn.rollback();
                } catch (IOException e) {
                    e.printStackTrace();
                } finally {
                    if (bufferedReader != null){
                        try {
                            bufferedReader.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }
                conn.close();
                synchronized (parent) {
                    parent.totalRows += recordsRead;
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }


    private static class TestDB {
        public static void main(String[] args) {
            new TPCHLoaderMySQL(1).createDatabase();
        }
    }

    private static class TestLoadDBL {
        public static void main(String[] args) {
            InputStream configStream = ConnectionMySQL.class.getClassLoader().getResourceAsStream("config.json");
            Configuration config = new Gson().fromJson(new InputStreamReader(configStream), Configuration.class);
            DBConnection DBconnect = new ConnectionMySQL(config);
            TPCHLoaderMySQL tpchLoaderMySQL = new TPCHLoaderMySQL(1);
            tpchLoaderMySQL.loadData(DBconnect);
        }
    }
}
